package cn.doitedu.demo2;

import cn.doitedu.demo1.EventBean;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/11/18
 * @Desc: 学大数据，上多易教育
 * -- 营销策略1：营销规则
 *    针对年龄在 20~28 之间，性别为 male的用户，                         -- 静态画像条件
 *    当他发生一个行为：添加购物车 时，即给他推送一个满200-50的优惠活动信息   -- 触发条件
 *
 **/
public class RuleCalculatorV2_Rule1 implements RuleCalculatorV2{

    RestHighLevelClient client;
    SearchRequest request;
    JSONObject message;

    @Override
    public void init(RuntimeContext runtimeContext) {
        // es的请求客户端
        client = new RestHighLevelClient(
                RestClient
                        .builder(new HttpHost("doitedu", 9200, "http"))
        );
        // 用于查询参数封装的对象
        request = new SearchRequest("doit42_userprofile");


        message = new JSONObject();
        message.put("rule_id", "rule-001");
    }

    @Override
    public void calculate(EventBean eventBean, Collector<String> out) throws IOException {
        if ("add_cart".equals(eventBean.getEvent_id())) {

            // 判断当前行为人，是否满足营销规则中的画像条件
            IdsQueryBuilder idQuery = QueryBuilders.idsQuery().addIds(eventBean.getUser_id() + "");
            RangeQueryBuilder ageQuery = QueryBuilders.rangeQuery("tag0202").gte(36).lte(40);
            MatchQueryBuilder genderQuery = QueryBuilders.matchQuery("tag0201", "male");

            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
            boolQueryBuilder.must(idQuery)
                    .must(ageQuery)
                    .must(genderQuery);


            request.source(new SearchSourceBuilder().query(boolQueryBuilder));

            // 用客户端发送请求
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            if (response.getHits().getHits().length > 0) {
                // 如果满足，则发送消息
                message.put("user_id", eventBean.getUser_id());
                message.put("hit_time", eventBean.getAction_time());
                out.collect(message.toJSONString());
            }
        }
    }
}
